/**
 * Copyright 2013 Netflix, Inc.
 * 
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 * 
 * http://www.apache.org/licenses/LICENSE-2.0
 * 
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package com.atom.zx.operators;

import com.atom.zx.Observer;
import com.atom.zx.plugins.RxJavaPlugins;
import com.atom.zx.util.CompositeException;
import com.atom.zx.util.OnErrorNotImplementedException;

import java.util.Arrays;
import java.util.concurrent.atomic.AtomicBoolean;


public class SafeObserver<T> implements Observer<T> {

    private final Observer<T> actual;
    private final AtomicBoolean isFinished = new AtomicBoolean(false);
    private final SafeObservableSubscription subscription;

    public SafeObserver(SafeObservableSubscription subscription, Observer<T> actual) {
        this.subscription = subscription;
        this.actual = actual;
    }

    @Override
    public void onCompleted() {
        if (isFinished.compareAndSet(false, true)) {
            try {
                actual.onCompleted();
            } catch (Throwable e) {
                // handle errors if the onCompleted implementation fails, not just if the Observable fails
                onError(e);
            }
            // auto-unsubscribe
            subscription.unsubscribe();
        }
    }

    @Override
    public void onError(Throwable e) {
        if (isFinished.compareAndSet(false, true)) {
            try {
                actual.onError(e);
            } catch (Throwable e2) {
                if (e2 instanceof OnErrorNotImplementedException) {
                    throw (OnErrorNotImplementedException) e2;
                } else {
                    RxJavaPlugins.getInstance().getErrorHandler().handleError(e);
                    RxJavaPlugins.getInstance().getErrorHandler().handleError(e2);
                    throw new RuntimeException("Error occurred when trying to propagate error to Observer.onError", new CompositeException(Arrays.asList(e, e2)));
                }
            }
            // auto-unsubscribe
            subscription.unsubscribe();
        }
    }

    @Override
    public void onNext(T args) {
        try {
            if (!isFinished.get()) {
                actual.onNext(args);
            }
        } catch (Throwable e) {
            onError(e);
        }
    }

}
